package com.twoandone.tms.realtime.app.dwd;

import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTransportTaskDetail {
    public static void main(String[] args) {
        // DwdTransportTaskDetail
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(4);

        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_transport_task_detail"));
//        tableEnv.executeSql("select * from topic_db").print();
//        从kafka表拿出运输任务线路明细表
        Table transportPlanLineDetail = tableEnv.sqlQuery("select \n" +
                "\t`data`['id'] transport_line_id," +
                "\t`data`['order_id'] order_id," +
                "\t`data`['start_org_id'] start_org_id," +
                "\t`data`['end_org_id'] end_org_id," +
                "\t`data`['line_base_id'] line_base_id " +
                " from `topic_db` \n" +
                "where `table` = 'transport_plan_line_detail'\n");
        tableEnv.createTemporaryView("transport_plan_line_detail", transportPlanLineDetail);

//        运输任务表
        Table transportTask = tableEnv.sqlQuery("select \n" +
                "\t`data`['id'] id,\n" +
                "\t`data`['shift_id'] shift_id,\n" +
                "\t`data`['line_id'] line_id,\n" +
                "\t`data`['start_org_id'] start_org_id,\n" +
                "\t`data`['start_org_name'] start_org_name,\n" +
                "\t`data`['end_org_id'] end_org_id,\n" +
                "\t`data`['end_org_name'] end_org_name,\n" +
                "\t`data`['status'] status,\n" +
                "\t`data`['order_num'] order_num,\n" +
                "\t`data`['driver1_emp_id'] driver1_emp_id,\n" +
                "\t`data`['driver1_name'] driver1_name,\n" +
                "\t`data`['driver2_emp_id'] driver2_emp_id,\n" +
                "\t`data`['driver2_name'] driver2_name, \n" +
                "\t`data`['truck_id'] truck_id,\n" +
                "\t`data`['truck_no'] truck_no, \n" +
                "\t`data`['actual_start_time'] actual_start_time,\n" +
                "\t`data`['actual_end_time'] actual_end_time,\n" +
                "\t`data`['actual_distance'] actual_distance,\n" +
                "\t`data`['create_time'] create_time,\n" +
                "\tts,\n" +
                "\tproc_time\n" +
                "from `topic_db`\n" +
                "where `table` = 'transport_task'\n");
        tableEnv.createTemporaryView("transport_task", transportTask);

//        运单运输关联表
        Table transportTaskDetail = tableEnv.sqlQuery("select\n" +
                "\t`data`['transport_task_id'] transport_task_id,\n" +
                "\t`data`['order_id'] order_id,\n" +
                "\t`data`['sorter_emp_id'] sorter_emp_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'transport_task_detail'\n");
        tableEnv.createTemporaryView("transport_task_detail", transportTaskDetail);

        tableEnv.executeSql(MySqlUtil.getBaseDicLookUpDDL());
//        关联
        Table joinTable = tableEnv.sqlQuery("select\n" +
                " task_detail.transport_task_id,\n" +
                " task.shift_id,\n" +
                " task.line_id,\n" +
                " task.start_org_id,\n" +
                " task.start_org_name,\n" +
                " task.end_org_id,\n" +
                " task.end_org_name,\n" +
                " task.status,\n" +
                " dic.name,\n" +
                " cast(dic.parent_id as string) ,\n" +
                " task.order_num,\n" +
                " task.driver1_emp_id,\n" +
                " task.driver1_name,\n" +
                " task.driver2_emp_id,\n" +
                " task.driver2_name,\n" +
                " task.truck_id,\n" +
                " task.truck_no,\n" +
                " task.actual_start_time,\n" +
                " task.actual_end_time,\n" +
                " task.actual_distance,\n" +
                " task.create_time,\n" +
                " task.ts,\n" +
                " task_detail.order_id,\n" +
                " task_detail.sorter_emp_id,\n" +
                " line_detail.line_base_id\n" +
                " from transport_task  task\n" +
                " join transport_task_detail task_detail\n" +
                " on task.id = task_detail.transport_task_id\n" +
                " join transport_plan_line_detail line_detail\n" +
                " on task_detail.order_id = line_detail.order_id\n" +
                " join base_dic  FOR SYSTEM_TIME AS OF task.proc_time AS dic\n" +
                "     ON cast(task.status as bigint) = dic.id ");

        tableEnv.createTemporaryView("joinTable", joinTable);

//        tableEnv.executeSql("select * from joinTable").print();

        tableEnv.executeSql("CREATE TABLE dwd_transport_task_detail ( " +
                "\ttransport_task_id string,\n" +
                "\tshift_id string,\n" +
                "\tline_id string,\n" +
                "\tstart_org_id string,\n" +
                "\tstart_org_name string,\n" +
                "\tend_org_id string,\n" +
                "\tend_org_name string,\n" +
                "\tstatus string,\n" +
                "\tname string,\n" +
                "\tparent_id string,\n" +
                "\torder_num string,\n" +
                "\tdriver1_emp_id string,\n" +
                "\tdriver1_name string,\n" +
                "\tdriver2_emp_id string,\n" +
                "\tdriver2_name string,\n" +
                "\ttruck_id string,\n" +
                "\ttruck_no string,\n" +
                "\tactual_start_time string,\n" +
                "\tactual_end_time string,\n" +
                "\tactual_distance string,\n" +
                "\tcreate_time string,\n" +
                "\tts string,\n" +
                "\torder_id string,\n" +
                "\tsorter_emp_id string,\n" +
                "\tline_base_id string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_transport_task_detail"));

        tableEnv.executeSql("insert into dwd_transport_task_detail select * from joinTable");


    }
}
